package com.maxflow.hadoop;

import java.io.IOException;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

public class Test {

	public static void main(String[] args) throws IOException {
		JobClient client = new JobClient();
		JobConf conf = new JobConf(com.maxflow.hadoop.Test.class);

		// TODO: specify output types
		conf.setOutputKeyClass(Text.class);
		conf.setOutputValueClass(IntWritable.class);

		// TODO: specify input and output DIRECTORIES (not files)
		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		// TODO : THIS PART MUST BE EDITED
		FileInputFormat.setInputPaths(conf, new Path("hdfs://mustafa-laptop:54310/user/books"));
		FileOutputFormat.setOutputPath(conf, new Path("hdfs://mustafa-laptop:54310/user/output"));

		// TODO: specify a mapper
		conf.setMapperClass(MapperClass.class);

		// TODO: specify a reducer
		conf.setReducerClass(ReducerClass.class);

		client.setConf(conf);

		try {
			JobClient.runJob(conf);
		} catch (Exception e) {
			e.printStackTrace();
		}

		Log log = LogFactory.getLog(Test.class);
		log.info(ReducerClass.text + " : " + ReducerClass.max);

		System.out.println(ReducerClass.text + " : " + ReducerClass.max);
	}
}
